package com.leyou.starter.es.repository;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.leyou.starter.es.anotations.Id;
import com.leyou.starter.es.anotations.Index;
import com.leyou.starter.es.dto.PageInfo;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.SuggestBuilders;
import org.elasticsearch.search.suggest.completion.CompletionSuggestion;
import org.springframework.util.CollectionUtils;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author 虎哥
 */
public class RepositoryHandler<T, ID> implements Repository<T, ID> {

    private final RestHighLevelClient client;
    private String indexName;
    private Class<T> tClass;
    private Class<ID> idClass;
    private Field idField;

    public RepositoryHandler(RestHighLevelClient client, Class<? extends Repository> interfaceClass) {
        this.client = client;

        // 利用反射，获取接口上的泛型的类型
        Type[] genericInterfaces = interfaceClass.getGenericInterfaces();
        if(genericInterfaces.length < 1){
            throw new RuntimeException();
        }
        // 泛型接口
        ParameterizedType genericInterface = (ParameterizedType) genericInterfaces[0];
        // 泛型接口中的泛型的类型
        Type[] actualTypeArguments = genericInterface.getActualTypeArguments();
        // 获取第一个泛型，实体类T的类型
        tClass = (Class<T>) actualTypeArguments[0];
        // 获取第二个泛型，实体类的ID的类型
        idClass = (Class<ID>) actualTypeArguments[1];

        // 利用反射获取注解
        if (tClass.isAnnotationPresent(Index.class)) {
            // 有, 取注解中的值作为索引库名
            Index index = tClass.getAnnotation(Index.class);
            indexName = index.value();
        }else{
            // 没有，用类名首字母小写作为索引库名
            String simpleName = tClass.getSimpleName();
            indexName = simpleName.substring(0, 1).toLowerCase() + simpleName.substring(1);
        }

        // 获取id
        Field[] declaredFields = tClass.getDeclaredFields();
        for (Field declaredField : declaredFields) {
            if (declaredField.isAnnotationPresent(Id.class)) {
                // 找到了id字段了
                idField = declaredField;
                declaredField.setAccessible(true);
            }
        }
    }

    private final static ObjectMapper mapper = new ObjectMapper();

    @Override
    public boolean createIndex(String source) {
        try {
            // 创建request
            CreateIndexRequest request = new CreateIndexRequest(indexName);
            // settings和mappings
            request.source(source, XContentType.JSON);
            // 发出请求
            CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
            // 返回结果
            return response.isAcknowledged();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public boolean deleteIndex() {
        try {
            // 创建request
            DeleteIndexRequest request = new DeleteIndexRequest(indexName);
            // 删除
            AcknowledgedResponse response = client.indices().delete(request, RequestOptions.DEFAULT);
            // 返回结果
            return response.isAcknowledged();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public boolean save(T t) {
        try {
            // 创建一个request对象
            IndexRequest request = new IndexRequest(indexName);
            // 指定id
            request.id(getId(t));
            // 指定文档
            request.source(toJson(t), XContentType.JSON);

            // 新增或修改
            IndexResponse response = client.index(request, RequestOptions.DEFAULT);

            // 判断是否成功
            return response.getShardInfo().getFailed() == 0;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    @Override
    public boolean deleteById(ID id) {
        try {
            // 创建request
            DeleteRequest request = new DeleteRequest(indexName, id.toString());
            // 删除
            DeleteResponse response = client.delete(request, RequestOptions.DEFAULT);
            // 判断是否成功
            return response.getShardInfo().getFailed() == 0;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public Mono<T> queryById(ID id) {
        GetRequest request = new GetRequest(indexName, id.toString());
        return Mono.create(sink -> {
            client.getAsync(request, RequestOptions.DEFAULT, new ActionListener<GetResponse>() {
                @Override
                public void onResponse(GetResponse response) {
                    if(!response.isExists()){
                        // 不存在
                        sink.error(new RuntimeException("数据不存在！"));
                        return;
                    }
                    // 查询成功，获取source
                    String json = response.getSourceAsString();
                    // 反序列化
                    T t = fromJson(json);
                    // 发射数据
                    sink.success(t);
                }

                @Override
                public void onFailure(Exception e) {
                    sink.error(e);
                }
            });
        });
    }

    @Override
    public boolean saveAll(Iterable<T> list) {
        try {
            // 创建批处理的请求
            BulkRequest request = new BulkRequest(indexName);
            // 循环添加 新增文档的请求
            for (T t : list) {
                request.add(new IndexRequest().id(getId(t)).source(toJson(t), XContentType.JSON));
            }
            // 发请求
            BulkResponse response = client.bulk(request, RequestOptions.DEFAULT);
            // 返回结果
            return response.hasFailures();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public Mono<PageInfo<T>> search(SearchSourceBuilder sourceBuilder) {
        // 创建搜索request
        SearchRequest request = new SearchRequest(indexName);
        // 添加搜索条件
        request.source(sourceBuilder);

        // 发出异步请求
        return Mono.create(sink -> {
            client.searchAsync(request, RequestOptions.DEFAULT, new ActionListener<SearchResponse>() {
                @Override
                public void onResponse(SearchResponse response) {
                    // 解析响应，得到分页结果
                    PageInfo<T> info = getInfoFromResponse(response);
                    // 发射
                    sink.success(info);
                }

                @Override
                public void onFailure(Exception e) {
                    sink.error(e);
                }
            });
        });
    }

    private PageInfo<T> getInfoFromResponse(SearchResponse response) {
        // 解析
        SearchHits searchHits = response.getHits();
        // 总条数
        long total = searchHits.getTotalHits().value;
        // 数据
        SearchHit[] hits = searchHits.getHits();
        // 最终的数据集合
        List<T> list = new ArrayList<>(hits.length);
        // 数据处理
        for (SearchHit hit : hits) {
            // 获取source
            String json = hit.getSourceAsString();
            // 反序列化
            T t = fromJson(json);
            list.add(t);
            // 获取高亮
            Map<String, HighlightField> map = hit.getHighlightFields();
            if(!CollectionUtils.isEmpty(map)){
                // 处理高亮
                for (HighlightField highlightField : map.values()) {
                    // 字段名称
                    String fieldName = highlightField.getName();
                    Text[] fragments = highlightField.getFragments();
                    String fieldValue = StringUtils.join(fragments);
                    // 反射设置值
                    try {
                        BeanUtils.setProperty(t, fieldName, fieldValue);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        }
        // 封装最终结果
        return new PageInfo<>(total, list);
    }

    @Override
    public Mono<List<String>> suggestBySingleField(String field, String prefix) {
        // 准备请求条件
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.suggest(new SuggestBuilder()
                .addSuggestion("xx",
                        SuggestBuilders.completionSuggestion(field).prefix(prefix).size(20)));
        // 准备请求对象
        SearchRequest request = new SearchRequest(indexName);
        request.source(sourceBuilder);

        return Mono.create(sink -> {
            // 发起请求
            client.searchAsync(request, RequestOptions.DEFAULT, new ActionListener<SearchResponse>() {
                @Override
                public void onResponse(SearchResponse response) {
                    // 获取suggest结果
                    Suggest suggest = response.getSuggest();
                    // 根据名称，获取suggestion
                    CompletionSuggestion suggestion = suggest.getSuggestion("xx");
                    // 获取options
                    List<String> list = suggestion.getOptions()
                            .stream()
                            .map(option -> option.getText())
                            .map(Text::string)
                            .collect(Collectors.toList());
                    // 发射
                    sink.success(list);
                }

                @Override
                public void onFailure(Exception e) {
                    sink.error(e);
                }
            });
        });
    }

    private String getId(T t) {
        try {
            return idField.get(t).toString();
        } catch (IllegalAccessException e) {
            throw new RuntimeException(e);
        }
    }

    private String toJson(T t) {
        try {
            return mapper.writeValueAsString(t);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }

    private T fromJson(String json) {
        try {
            return mapper.readValue(json, tClass);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
